// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: heartbeat optimize
// feature: Support Cache fast application and slow release mechanism
// feature: Split history as an independent common module
// feature: discovery support client and server
// feature: add statictic lookup
// feature: Crop code for in-process communication
// ------------------------------------------------------------------

#ifndef INCLUDE_EDDS_RTPS_WRITER_STATEFULWRITER_H_
#define INCLUDE_EDDS_RTPS_WRITER_STATEFULWRITER_H_

#ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC

#include <edds/rtps/writer/RTPSWriter.h>
#include <history/IChangePool.h>
#include <history/IPayloadPool.h>
#include <condition_variable>
#include <mutex>
#include <ertps/utils/collections/ResourceLimitedVector.hpp>
#include <edds/rtps/interfaces/IReaderDataFilter.hpp>

namespace vbs {
namespace common {
class TimedEvent;

}
}  // namespace vbs

namespace evbs {
namespace ertps {
namespace rtps {
using vbs::common::TimedEvent;
class ReaderProxy;
/**
 * Class StatefulWriter, specialization of RTPSWriter that maintains information of each matched
 * Reader.
 * @ingroup WRITER_MODULE
 */
class StatefulWriter : public RTPSWriter {
    friend class RTPSParticipantImpl;
    friend class ReaderProxy;

 public:
    //! Destructor
    virtual ~StatefulWriter();

 protected:
    //! Constructor
    StatefulWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                   edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    StatefulWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                   const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
                   edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    StatefulWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                   const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
                   const std::shared_ptr<vbs::common::IChangePool>& change_pool,
                   edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    void rebuild_status_after_load();

    virtual void print_inconsistent_acknack(const GUID_t& writer_guid, const GUID_t& reader_guid,
                                            const SequenceNumber_t& min_requested_sequence_number,
                                            const SequenceNumber_t& max_requested_sequence_number,
                                            const SequenceNumber_t& next_sequence_number);

 private:
    void init(RTPSParticipantImpl* pimpl, const WriterAttributes& att);

    //! Timed Event to manage the periodic HB to the Reader.
    vbs::common::TimedEvent* periodic_hb_event_;

    //! Timed Event to manage the Acknack response delay.
    vbs::common::TimedEvent* nack_response_event_;

    //! A timed event to mark samples as acknowledget (used only if disable positive ACKs QoS is
    //! enabled)
    vbs::common::TimedEvent* ack_event_;

    //! Count of the sent heartbeats.
    Count_t m_heartbeatCount;
    //! WriterTimes
    WriterTimes m_times;

    //! Vector containing all the remote ReaderProxies.
    ResourceLimitedVector<ReaderProxy*> matched_remote_readers_;
    //! Vector containing all the inactive, ready for reuse, ReaderProxies.
    ResourceLimitedVector<ReaderProxy*> matched_readers_pool_;

    //! To avoid notifying twice of the same sequence number
    SequenceNumber_t next_all_acked_notify_sequence_;
    SequenceNumber_t min_readers_low_mark_;

    // TODO Join this mutex when main mutex would not be recursive.
    std::mutex all_acked_mutex_;
    std::condition_variable all_acked_cond_;
    // TODO Also remove when main mutex not recursive.
    bool all_acked_;
    std::condition_variable_any may_remove_change_cond_;
    uint32_t may_remove_change_;
    vbs::common::TimedEvent* free_event_ = nullptr;

 public:
    void acknack_request() override;

    /**
     * Add a specific change to all ReaderLocators.
     * @param p Pointer to the change.
     * @param max_blocking_time
     */
    void unsent_change_added_to_history(CacheChange_t* change,
                                        const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time,
                                        bool in_history) override;

    /**
     * Indicate the writer that a change has been removed by the history due to some HistoryQos
     * requirement.
     * @param a_change Pointer to the change that is going to be removed.
     * @return True if removed correctly.
     */
    bool change_removed_by_history(CacheChange_t* const change) override;

    /**
     * Sends a change directly to a intraprocess reader.
     */
    bool intraprocess_delivery(CacheChange_t* change, ReaderProxy* reader_proxy);

    bool intraprocess_gap(ReaderProxy* reader_proxy, const SequenceNumber_t& first_seq,
                          const SequenceNumber_t& last_seq);

    bool intraprocess_gap(ReaderProxy* const reader_proxy, const SequenceNumber_t& seq_num) {
        const SequenceNumber_t last_seq = seq_num + 1U;
        return intraprocess_gap(reader_proxy, seq_num, last_seq);
    }

    bool intraprocess_heartbeat(ReaderProxy* reader_proxy, bool liveliness = false);

    //! Increment the HB count.
    inline void incrementHBCount() { ++m_heartbeatCount; }

    template <typename Function>
    Function for_each_reader_proxy(Function f) const {
        // we cannot directly pass iterators neither const_iterators to matched_readers_ because then the functor would
        // be able to modify ReaderProxy elements
        for (const ReaderProxy* rp : matched_local_readers_) {
            f(rp);
        }
        for (const ReaderProxy* rp : matched_remote_readers_) {
            f(rp);
        }

        return f;
    }

    /**
     * Add a matched reader.
     * @param data Pointer to the ReaderProxyData object added.
     * @return True if added.
     */
    bool matched_reader_add(ReaderProxyData& data) override;

    /**
     * Remove a matched reader.
     * @param reader_guid GUID of the reader to remove.
     * @return True if removed.
     */
    bool matched_reader_remove(const GUID_t& reader_guid) override;

    /**
     * Tells us if a specific Reader is matched against this writer
     * @param reader_guid GUID of the reader to check.
     * @return True if it was matched.
     */
    bool matched_reader_is_matched(const GUID_t& reader_guid) override;

    /**
     * @brief Check if a specific change has been delivered to the transport layer at least once for every matched
     * remote RTPSReader.
     *
     * @param seq_num Sequence number of the change to check.
     * @return true if delivered.
     * @return false otherwise.
     */
    bool has_been_fully_delivered(const SequenceNumber_t& seq_num) const override;

    bool is_acked_by_all(const CacheChange_t* a_change) const override;
    bool is_acked_by_all(const SequenceNumber_t seq) const;

    bool wait_for_all_acked(const Duration_t& max_wait) override;

    bool all_readers_updated();

    /**
     * Remove the change with the minimum SequenceNumber
     * @return True if removed.
     */
    bool try_remove_change(const std::chrono::steady_clock::time_point& max_blocking_time_point,
                           std::unique_lock<vbs::common::RecursiveTimedMutex>& lock) override;

    bool wait_for_acknowledgement(const SequenceNumber_t& seq,
                                  const std::chrono::steady_clock::time_point& max_blocking_time_point,
                                  std::unique_lock<vbs::common::RecursiveTimedMutex>& lock) override;

    /**
     * Update the Attributes of the Writer.
     * @param att New attributes
     */
    void updateAttributes(const WriterAttributes& att) override;

    /**
     * Find a Reader Proxy in this writer.
     * @param[in] readerGuid The GUID_t of the reader.
     * @param[out] RP Pointer to pointer to return the ReaderProxy.
     * @return True if correct.
     */
    bool matched_reader_lookup(GUID_t& readerGuid, ReaderProxy** RP);

    /** Get count of heartbeats
     * @return count of heartbeats
     */
    inline Count_t getHeartbeatCount() const { return this->m_heartbeatCount; }

    /**
     * Get the RTPS participant
     * @return RTPS participant
     */
    inline RTPSParticipantImpl* getRTPSParticipant() const { return mp_RTPSParticipant; }

    /**
     * Get the number of matched readers
     * @return Number of the matched readers
     */
    size_t getMatchedReadersSize() const override;

    /**
     * @brief Returns true if disable positive ACKs QoS is enabled
     *
     * @return True if positive acks are disabled, false otherwise
     */
    inline bool get_disable_positive_acks() const override { return disable_positive_acks_; }

    /**
     * Update the WriterTimes attributes of all associated ReaderProxy.
     * @param times WriterTimes parameter.
     */
    void updateTimes(const WriterTimes& times);

    SequenceNumber_t next_sequence_number() const;

    /**
     * @brief Sends a periodic heartbeat
     *
     * @param final Final flag
     * @param liveliness Liveliness flag
     *
     * @return True on success
     */
    bool send_periodic_heartbeat(bool final = false, bool liveliness = false);

    /*!
     * @brief Sends a heartbeat to a remote reader.
     *
     * @remarks This function is non thread-safe.
     */
    bool send_heartbeat_to_nts(ReaderProxy& remoteReaderProxy, bool liveliness = false, bool force = false);

    void perform_nack_response();

    void perform_nack_supression(const GUID_t& reader_guid);

    /**
     * Process an incoming ACKNACK submessage.
     * @param[in] writer_guid      GUID of the writer the submessage is directed to.
     * @param[in] reader_guid      GUID of the reader originating the submessage.
     * @param[in] ack_count        Count field of the submessage.
     * @param[in] sn_set           Sequence number bitmap field of the submessage.
     * @param[in] final_flag       Final flag field of the submessage.
     * @param[out] result          true if the writer could process the submessage.
     *                             Only valid when returned value is true.
     * @return true when the submessage was destinated to this writer, false otherwise.
     */
    bool process_acknack(const GUID_t& writer_guid, const GUID_t& reader_guid, uint32_t ack_count,
                         const SequenceNumberSet_t& sn_set, const bool final_flag, bool& result) override;

    /**
     * Process an incoming NACKFRAG submessage.
     * @param[in] writer_guid      GUID of the writer the submessage is directed to.
     * @param[in] reader_guid      GUID of the reader originating the submessage.
     * @param[in] ack_count        Count field of the submessage.
     * @param[in] seq_num          Sequence number field of the submessage.
     * @param[in] fragments_state  Sequence number field of the submessage.
     * @param[out] result          true if the writer could process the submessage.
     *                             Only valid when returned value is true.
     * @return true when the submessage was destinated to this writer, false otherwise.
     */
    bool process_nack_frag(const GUID_t& writer_guid, const GUID_t& reader_guid, uint32_t ack_count,
                           const SequenceNumber_t& seq_num, const FragmentNumberSet_t fragments_state,
                           bool& result) override;

    /*!
     * Tells writer the sample can be sent to the network.
     * This function should be used by a edds::rtps::FlowController.
     *
     * @param cache_change Pointer to the CacheChange_t that represents the sample which can be
     * sent.
     * @param group RTPSMessageGroup reference uses for generating the RTPS message.
     * @param locator_selector RTPSMessageSenderInterface reference uses for selecting locators. The
     * reference has to be a member of this RTPSWriter object.
     * @param max_blocking_time Future timepoint where blocking send should end.
     * @return Return code.
     * @note Must be non-thread safe.
     */
    DeliveryRetCode deliver_sample_nts(CacheChange_t* cache_change, RTPSMessageGroup& group,
                                       LocatorSelectorSender& locator_selector,
                                       const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time,
                                       bool in_history = true) override;

    LocatorSelectorSender& get_general_locator_selector() override { return locator_selector_general_; }

    LocatorSelectorSender& get_async_locator_selector() override { return locator_selector_async_; }

    inline void setSamplesPerHb(int value) { samples_per_hb = value; }

    edds::dds::builtin::StatisticMatchGuids get_remote_guids() override;

    edds::dds::builtin::StatisticProxyInfos get_proxy_infos() override;

    void FilterReaderLocators(ReaderProxyData& rdata, ReaderProxy* reader);

 private:
    void update_reader_info(LocatorSelectorSender& locator_selector, bool create_sender_resources);

    void select_all_readers_nts(RTPSMessageGroup& group, LocatorSelectorSender& locator_selector);

    void send_heartbeat_piggyback_nts_(RTPSMessageGroup& message_group, LocatorSelectorSender& locator_selector,
                                       uint32_t& last_bytes_processed);

    void send_heartbeat_nts_(size_t number_of_readers, RTPSMessageGroup& message_group, bool final,
                             bool liveliness = false, ReaderProxy* reader = nullptr);

    void check_acked_status();

    /**
     * @brief A method called when the ack timer expires
     *
     * @details Only used if disable positive ACKs QoS is enabled
     */
    bool ack_timer_expired();

    void deliver_sample_to_intraprocesses(CacheChange_t* change);

    DeliveryRetCode deliver_sample_to_network(
        CacheChange_t* change, RTPSMessageGroup& group, LocatorSelectorSender& locator_selector,
        const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time);

    /**
     * Check the StatefulWriter's sequence numbers and add the required GAP messages to the provided
     * message group.
     *
     * @param group     Reference to the Message Group to which the GAP messages are to be added.
     */
    void add_gaps_for_holes_in_history_(RTPSMessageGroup& group);

    //! True to disable piggyback heartbeats
    bool disable_heartbeat_piggyback_;
    //! True to disable positive ACKs
    bool disable_positive_acks_;
    //! Keep duration for disable positive ACKs QoS, in microseconds
    std::chrono::duration<double_t, std::ratio<1, 1000000>> keep_duration_us_;
    //! Last acknowledged cache change (only used if using disable positive ACKs QoS)
    SequenceNumber_t last_sequence_number_;
    //! Biggest sequence number removed from history
    SequenceNumber_t biggest_removed_sequence_number_;

    const uint32_t sendBufferSize_;

    int32_t currentUsageSendBufferSize_;

    bool there_are_remote_readers_ = false;

    bool there_are_local_readers_ = false;

    StatefulWriter& operator=(const StatefulWriter&) = delete;

    //! Vector containing all the active ReaderProxies for intraprocess delivery.
    ResourceLimitedVector<ReaderProxy*> matched_local_readers_;

    LocatorSelectorSender locator_selector_general_;

    LocatorSelectorSender locator_selector_async_;

    uint32_t samples_per_hb;

    uint32_t piggyback_sample_count = 0U;

    bool fast_heartbeat_ = false;
    //When unack sample counts exceeds high_watermark, change to fast heartbeat period.
    int32_t high_watermark = -1;
    //When unack sample counts fall below high_watermark, change to heartbeat period.
    int32_t low_watermark = -1;
    uint32_t mem_watermark = 0;
};

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */

#endif  // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
#endif  // INCLUDE_EDDS_RTPS_WRITER_STATEFULWRITER_H_
